package com.example.core.queue.distuptor.producer;

import com.alibaba.fastjson.JSON;
import com.example.core.queue.distuptor.config.BufferData;
import com.example.core.queue.distuptor.config.DisruptorConfig;
import com.example.core.queue.distuptor.event.QueueEvent;
import com.lmax.disruptor.RingBuffer;
import org.springframework.beans.BeanUtils;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;

/**
 * 队列生产者
 */
public  class QueueProducer {

    private String queueId;
    private final RingBuffer<QueueEvent> ringBuffer;

    public QueueProducer(){
        this.queueId= DisruptorConfig.DEFULTDISRUPTORID;
        ringBuffer= BufferData.init().get(queueId);
    }
    public QueueProducer(String queueId){
        this.queueId=queueId;
        ringBuffer=BufferData.init().get(queueId);
    }


    /**
     * 向队列中写入数据
     * @param data
     */
    public void sendData(QueueEvent data) {
        String jsonObject= JSON.toJSONString(data);
        byte[] body = jsonObject.getBytes();
        int bodyLen = 0;
        if (body != null) {
            bodyLen = body.length;
        }
        int allLen = DisruptorConfig.HEADER_LENGHT + bodyLen;
        ByteBuffer byteBuffer = ByteBuffer.allocate(allLen);
        byteBuffer.order(ByteOrder.BIG_ENDIAN);
        byteBuffer.putInt(bodyLen);
        if (body != null) {
            byteBuffer.put(body);
        }
        ringBuffer.publishEvent((event, sequence, buffer) -> {
            buffer.flip();//读模式
            byte[] bytes=new byte[buffer.getInt()];
            buffer.get(bytes);
            String str=new String(bytes);
            QueueEvent entity=JSON.parseObject(str,event.getClass());
            BeanUtils.copyProperties(entity,event);
        }, byteBuffer);
    }




}
